package com.demo.rocketmq.api;

import cn.hutool.core.date.DateUtil;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Description
 * @Author likz
 * @Date 2024/9/10 10:54
 **/

@Slf4j
@RestController
@RequestMapping("api/test/dis")
@RequiredArgsConstructor
@Api(tags = "测试API-测试不同组消费")
public class TestDisGroupController {

    private final RocketMQTemplate rocketMQTemplate;

    public static final String TOPIC = "group";
    public static final String TOPIC_GROUP = "topic_group";
    // 主题+tag，中间用“:”分隔，主要是用于消息的过滤，比如说在消费的时候，只消费tag1标签下的消息
    private static final String TOPIC_TAG = TOPIC.concat(":tag1");
    private static final String TOPIC_TAG2 = TOPIC.concat(":tag2");
    private static final String TOPIC_GROUP_TAG = TOPIC_GROUP.concat(":tag1");
    private static final String TOPIC_GROUP_TAG2 = TOPIC_GROUP.concat(":tag2");

    @ApiOperation("标签过滤消息")
    @PostMapping("/hello")
    public String hello() {
        String msgContent = DateUtil.now()+"标签消息1";
        String msgContent2 = DateUtil.now()+"标签消息2";
        log.info("[生产者] 发送消息：{}", msgContent);
        this.rocketMQTemplate.convertAndSend(TOPIC_TAG, msgContent);
        this.rocketMQTemplate.convertAndSend(TOPIC_TAG2, msgContent2);
        return "OK";
    }

    @ApiOperation("标签过滤广播消息")
    @PostMapping("/hello2")
    public String hello2() {
        String msgContent = DateUtil.now()+"标签消息1";
        String msgContent2 = DateUtil.now()+"标签消息2";
        log.info("[生产者] 发送消息：{}", msgContent);
        this.rocketMQTemplate.convertAndSend(TOPIC_GROUP_TAG, msgContent);
        this.rocketMQTemplate.convertAndSend(TOPIC_GROUP_TAG2, msgContent2);
        return "OK";
    }


    @Slf4j
    @Service
    @RocketMQMessageListener(
            topic = TestDisGroupController.TOPIC, // 主题，指消费者组订阅的消息服务
//            selectorExpression = "tag1", // 选择哪个标签(tag)下的信息，默认是消费该主题下的所有信息
            consumerGroup = "dis-consumer-group", // 消费者组，一个组可以有多个消费者，主要的作用是集群模式负载均衡的实现，广播模式的通知的实现
            consumeMode = ConsumeMode.CONCURRENTLY, // 控制消费模式，你可以选择并发或有序接收消息
            messageModel = MessageModel.BROADCASTING // 控制消息模式，广播模式：所有消费者都能接收到信息，集群模式：无论有多少个消费者，只有一个消费者能够接收到信息，也就是说消息一旦被消费了，其它消费者就不能消费该条消息
    )
    static class BatchRocketMQListener implements RocketMQListener<String> {

        @Override
        public void onMessage(String msg) {
            log.info("[消费者] 接收全部消息：{}", msg);
        }

    }


    @Slf4j
    @Service
    @RocketMQMessageListener(
            topic = TestDisGroupController.TOPIC, // 主题，指消费者组订阅的消息服务
            selectorExpression = "tag1", // 选择哪个标签(tag)下的信息，默认是消费该主题下的所有信息
            consumerGroup = "dis-consumer-group2", // 消费者组，一个组可以有多个消费者，主要的作用是集群模式负载均衡的实现，广播模式的通知的实现
            consumeMode = ConsumeMode.CONCURRENTLY, // 控制消费模式，你可以选择并发或有序接收消息
            messageModel = MessageModel.BROADCASTING // 控制消息模式，广播模式：所有消费者都能接收到信息，集群模式：无论有多少个消费者，只有一个消费者能够接收到信息，也就是说消息一旦被消费了，其它消费者就不能消费该条消息
    )
    static class BatchRocketMQListener2 implements RocketMQListener<String> {

        @Override
        public void onMessage(String msg) {
            log.info("[消费者] 接收标签1消息：{}", msg);
        }

    }

    @Slf4j
    @Service
    @RocketMQMessageListener(
            topic = TestDisGroupController.TOPIC, // 主题，指消费者组订阅的消息服务
            selectorExpression = "tag2", // 选择哪个标签(tag)下的信息，默认是消费该主题下的所有信息
            consumerGroup = "dis-consumer-group3", // 消费者组，一个组可以有多个消费者，主要的作用是集群模式负载均衡的实现，广播模式的通知的实现
            consumeMode = ConsumeMode.CONCURRENTLY, // 控制消费模式，你可以选择并发或有序接收消息
            messageModel = MessageModel.BROADCASTING // 控制消息模式，广播模式：所有消费者都能接收到信息，集群模式：无论有多少个消费者，只有一个消费者能够接收到信息，也就是说消息一旦被消费了，其它消费者就不能消费该条消息
    )
    static class BatchRocketMQListener3 implements RocketMQListener<String> {

        @Override
        public void onMessage(String msg) {
            log.info("[消费者] 接收标签2消息：{}", msg);
        }

    }



}
